package com.lhd.app.processor;

import com.lhd.app.bean.PetTypeLabel;
import com.lhd.app.bean.UserBehavior;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.*;

public class PetTypeLabelProcessor extends ProcessWindowFunction<UserBehavior, PetTypeLabel, String, TimeWindow> {
    
    private static final Map<String, Double> BEHAVIOR_WEIGHTS = loadBehaviorWeights();
    
    @Override
    public void process(String userId, Context context, 
                      Iterable<UserBehavior> behaviors, 
                      Collector<PetTypeLabel> out) {
        
        Map<String, Double> petScores = new HashMap<>();
        
        for (UserBehavior behavior : behaviors) {
            // 获取宠物类型映射
            Set<String> petTypes = mapCategoryToPetTypes(
                behavior.getCategory1(), 
                behavior.getCategory2(), 
                behavior.getCategory3()
            );
            
            // 计算行为权重
            double weight = BEHAVIOR_WEIGHTS.getOrDefault(
                behavior.getBehaviorType(), 0.0);
            
            // 处理通用类目权重分配
            if (isGeneralCategory(behavior.getCategory3())) {
                weight = weight * 0.5; // 通用类目权重减半
            }
            
            // 累加分数
            for (String petType : petTypes) {
                petScores.merge(petType, weight, Double::sum);
            }
        }
        
        // 生成标签
        List<String> labels = generateLabels(petScores);
        
        PetTypeLabel result = new PetTypeLabel();
        result.setUserId(userId);
        result.setPetScores(petScores);
        result.setPetLabels(labels);
        result.setUpdateTime(System.currentTimeMillis());
        
        out.collect(result);
    }
    
    private Set<String> mapCategoryToPetTypes(String cat1, String cat2, String cat3) {
        Set<String> petTypes = new HashSet<>();
        
        // 根据类目映射规则确定宠物类型
        if ("狗狗".equals(cat1) || "狗零食".equals(cat1) || "犬主粮".equals(cat1)) {
            petTypes.add("狗");
        } else if ("猫咪".equals(cat1) || "猫零食".equals(cat1) || "猫主粮".equals(cat1)) {
            petTypes.add("猫");
        } else if ("仓鼠类及其它小宠".equals(cat1)) {
            // 根据子类目细分
            if (cat3 != null) {
                if (cat3.contains("仓鼠")) petTypes.add("仓鼠");
                if (cat3.contains("龙猫")) petTypes.add("龙猫");
                if (cat3.contains("宠物狐狸")) petTypes.add("宠物狐狸");
                if (cat3.contains("宠物貂")) petTypes.add("宠物貂");
                if (cat3.contains("香猪")) petTypes.add("香猪");
                if (cat3.contains("豚鼠")) petTypes.add("豚鼠");
            }
        } else if ("鸟类及用品".equals(cat1)) {
            petTypes.add("鸟类");
        } else if ("爬虫/鸣虫及其用品".equals(cat1)) {
            petTypes.add("爬虫/鸣虫");
        } else if ("水族世界".equals(cat1)) {
            petTypes.add("水族");
        } else if ("兔类及其用品".equals(cat1)) {
            petTypes.add("兔类");
        } else if ("马类及其用品".equals(cat1)) {
            petTypes.add("马类");
        }
        
        // 处理通用类目
        if (isGeneralCategory(cat3)) {
            petTypes.add("猫");
            petTypes.add("狗");
        }
        
        return petTypes;
    }
    
    private boolean isGeneralCategory(String category3) {
        Set<String> generalCategories = new HashSet<>(Arrays.asList(
            "猫/狗玩具", "猫/狗保健品", "猫/狗美容清洁用品", 
            "猫/狗日用品", "猫/狗医疗用品"
        ));
        return generalCategories.contains(category3);
    }
    
    private List<String> generateLabels(Map<String, Double> petScores) {
        List<String> labels = new ArrayList<>();
        for (Map.Entry<String, Double> entry : petScores.entrySet()) {
            if (entry.getValue() >= 10.0) { // 阈值判断
                labels.add(entry.getKey());
            }
        }
        return labels;
    }
    
    private static Map<String, Double> loadBehaviorWeights() {
        Map<String, Double> weights = new HashMap<>();
        weights.put("buy", 5.0);
        weights.put("cart", 3.0);
        weights.put("favorite", 2.0);
        weights.put("browse", 1.0);
        return weights;
    }
    
    // 判断是否为宠物相关行为
    public static boolean isPetRelatedBehavior(UserBehavior behavior) {
        Set<String> petCategories = new HashSet<>(Arrays.asList(
            "狗狗", "狗零食", "犬主粮", "猫咪", "猫零食", "猫主粮",
            "仓鼠类及其它小宠", "鸟类及用品", "爬虫/鸣虫及其用品", 
            "水族世界", "兔类及其用品", "马类及其用品"
        ));
        return petCategories.contains(behavior.getCategory1());
    }
}